package com.innovation.ic.sc.base.thread.listener.rabbitmq;

import com.innovation.ic.b1b.framework.manager.ThreadPoolManager;
import com.innovation.ic.sc.base.pojo.constant.handler.RabbitMqConstants;
import com.innovation.ic.sc.base.service.sc.AdvantageBrandService;
import com.innovation.ic.sc.base.service.sc.AdvantageModelService;
import com.rabbitmq.client.*;
import lombok.SneakyThrows;
import java.io.IOException;

/**
 * @desc   监听优势型号Erp通知队列
 * @author linuo
 * @time   2023年2月7日09:22:08
 */
public class ListenAdvantageModelNotifyQueueThread extends AbstractRabbitmqThread implements Runnable {
    private Channel channel;
    private AdvantageModelService advantageModelService;
    private AdvantageBrandService advantageBrandService;
    private ThreadPoolManager threadPoolManager;

    public ListenAdvantageModelNotifyQueueThread(Channel channel, AdvantageModelService advantageModelService, AdvantageBrandService advantageBrandService, ThreadPoolManager threadPoolManager) {
        this.channel = channel;
        this.advantageModelService = advantageModelService;
        this.advantageBrandService = advantageBrandService;
        this.threadPoolManager = threadPoolManager;
    }

    @Override
    @SneakyThrows
    public void run() {
        String exchange = RabbitMqConstants.SC_ERP_EXCHANGE;
        String routingKey = RabbitMqConstants.ERP_SC_ADD_NOTIFY_QUEUE;
        final String queue = routingKey;
        channel.queueBind(queue, exchange, routingKey);
        channel.queueDeclare(queue, true, false, false, null);
        channel.exchangeDeclare(exchange, RabbitMqConstants.DIRECT_TYPE, true);
        Consumer callback = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String bodyString = new String(body);
                HandleAdvantageDataThread handleAdvantageDataThread = new HandleAdvantageDataThread(queue, bodyString, advantageModelService, advantageBrandService);
                threadPoolManager.execute(handleAdvantageDataThread);
            }
        };
        channel.basicConsume(queue, true, callback);
    }
}